/*
* Copyright 2016 YONYOU Corporation Ltd. All Rights Reserved.
*
* This software is published under the terms of the YONYOU Software
* License version 1.0, a copy of which has been included with this
* distribution in the LICENSE.txt file.
*
* @Project         marketing

* @Package         com.yonyou.marketing.message.push.entrance

* @Author          LuZhen

* @FileName        MessageReceiver

* @Date            16/11/3

----------------------------------------------------------------------------------
*     Date          Who             Version         Comments
* 1. 16/11/3        LuZhen          1.0             Create Source
*
----------------------------------------------------------------------------------
*/


package com.yonyou.gmmc.service.wechatpush.mq.entrance;

import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.yonyou.gmmc.service.wechatpush.provider.PushThreadCallable;
import com.yonyou.gmmc.service.wechatpush.provider.PushThreadCallback;
import com.yonyou.gmmc.service.wechatpush.resolver.MessageResolver;
import com.yonyou.gmmc.service.wechatpush.resolver.PushMessage;

import org.apache.log4j.Logger;
import org.marketing.common.constants.MessagePushConstant;
import org.marketing.common.entity.wechatpushmessage.*;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;



@Component
public class MessageReceiver {

    private static Logger logger = Logger.getLogger(MessageReceiver.class);

    @Autowired
    private MessageResolver messageResolver;

    @Autowired
    private MessageSender messageSender;
    /**
     * bean 消费 test
     * @param content  消息内容 字符串
     */
    @RabbitListener(queues = MessagePushConstant.QUEUE_TEST,containerFactory="rabbitListenerContainerFactory")
    public void receiveTestMessage(@Payload String content) {
        logger.debug("start to listen");
        logger.debug("@@@@ test content is "+content);
        logger.debug("end to listen " );
    }


    @Autowired
    private ApplicationContext context;

    @Autowired
    private ListeningExecutorService executorService;




    /**
     * bean 消费
     * @param protocol  接口协议对象
     */
    @RabbitListener(queues = MessagePushConstant.WECHAT_QUEUE,containerFactory="rabbitListenerContainerFactory")
    public void listen(@Payload WeChatMessageProtocol protocol) {
        logger.debug("start to listen");
        logger.debug("protocol is "+protocol);
        //获取处理信息
        WeChatMessageProtocolHandleInfo newHandleInfo = WeChatMessageProtocolHandleInfo.init(protocol.getHandleInfo());
        protocol.setHandleInfo(newHandleInfo);
        //丢弃消息
        if(newHandleInfo.isDeprive()){
            logger.error("deprive message "+protocol);
            return;
        }
        //转换为协议消息
        PushMessage pushMessage=messageResolver.receiveMessage(protocol);
        //验证协议消息
        if(!messageResolver.validateMessage(pushMessage)){
            logger.error("message validate error");
            newHandleInfo.addHandleSimpleHistory("message validate error");
            protocol.setHandleInfo(newHandleInfo);
            messageSender.send(protocol);
            return;
        }
        //转入发送线程
        PushThreadCallable pushThreadCallable = context.getBean(MessagePushConstant.PROVIDER_THREAD,PushThreadCallable.class);
        pushThreadCallable.setProviderMessage(pushMessage);
        PushThreadCallback pushThreadCallback = context.getBean(PushThreadCallback.class,protocol);
        Futures.addCallback(executorService.submit(pushThreadCallable),pushThreadCallback);
        logger.debug("end to listen " );
    }

}
